package com.gzyj.flink.driver.function;

import com.alibaba.fastjson.JSON;
import com.gzyj.flink.driver.DriverInfo;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.driver.utils.TimeUtils;
import com.gzyj.flink.veh.AnbiaoVehicle;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author lhj
 */
public class VehInfoProcessFun extends BroadcastProcessFunction<String, AnbiaoVehicle, Driverlogininfo> {

    private MapStateDescriptor<String, AnbiaoVehicle> vehicleMapState;

    public VehInfoProcessFun(MapStateDescriptor<String, AnbiaoVehicle> vehicleMapState) {
        this.vehicleMapState = vehicleMapState;
    }

    @Override
    public void processElement(String s, BroadcastProcessFunction<String, AnbiaoVehicle, Driverlogininfo>.ReadOnlyContext readOnlyContext, Collector<Driverlogininfo> collector) throws Exception {

        DriverInfo driverInfo = JSON.parseObject(s, DriverInfo.class);

        ReadOnlyBroadcastState<String, AnbiaoVehicle> vehicleReadOnlyBroadcastState = readOnlyContext.getBroadcastState(this.vehicleMapState);

        if (StringUtils.isNotEmpty(driverInfo.getTerminalPhoneNo())
                && vehicleReadOnlyBroadcastState.contains(driverInfo.getTerminalPhoneNo())) {
            AnbiaoVehicle vehicle = vehicleReadOnlyBroadcastState.get(driverInfo.getTerminalPhoneNo());

            Driverlogininfo driverlogininfo = new Driverlogininfo();
            driverlogininfo.setVehId(vehicle.getId());
            driverlogininfo.setCheckObject(driverInfo.getIC_Card_Status().getCode());
            driverlogininfo.setLoginResults(driverInfo.getIC_Card_ReadResult().getCode());
            driverlogininfo.setDriverName(driverInfo.getDriverUserName());
            driverlogininfo.setDriverNo(StringUtils.isNotEmpty(driverInfo.getDriverIdentityCard()) ? driverInfo.getDriverIdentityCard() : driverInfo.getQualificationCode());
            driverlogininfo.setLoginTime(TimeUtils.toDate(driverInfo.getIC_Card_PlugDateTime() == null ? TimeUtils.getNow() : driverInfo.getIC_Card_PlugDateTime()));

            collector.collect(driverlogininfo);
        }


    }

    @Override
    public void processBroadcastElement(AnbiaoVehicle anbiaoVehicle, BroadcastProcessFunction<String, AnbiaoVehicle, Driverlogininfo>.Context context, Collector<Driverlogininfo> collector) throws Exception {
        BroadcastState<String, AnbiaoVehicle> bcState = context.getBroadcastState(vehicleMapState);
        bcState.put(anbiaoVehicle.getZongduanid(), anbiaoVehicle);
    }
}
